package com.lmh.interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;
import java.util.Map;

/**
 * @program flume-component
 * @Description 向header添加时间戳
 * @Author limeihua
 * @Date 2024/09/02 15:59
 * @Version 1.0
 */
public class TimeStampInterceptor implements Interceptor{

    public Integer timestampSerialNumber;
    public String separatorString;

    public TimeStampInterceptor(Integer timestampSerialNumber, String separator){
        this.timestampSerialNumber = timestampSerialNumber;
        this.separatorString = separator;
    }
    // 初始化操作
    @Override
    public void initialize() {

    }

    // 拦截
    @Override
    public Event intercept(Event event) {
        byte[] body = event.getBody();
        Map<String, String> headers = event.getHeaders();
        String bodyString = new String(body);
        String[] split = bodyString.split(separatorString);
        headers.put("timestamp", split[timestampSerialNumber]);
        event.setHeaders(headers);
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        for (Event event : list) {
            intercept(event);
        }
        return list;
    }

    @Override
    public void close() {

    }


    public static class Builder implements Interceptor.Builder{
        public Integer timestampSerialNumber;

        public String separatorString;

        @Override
        public Interceptor build() {
            return new TimeStampInterceptor(timestampSerialNumber, separatorString);
        }

        // 配置参数
        @Override
        public void configure(Context context) {
            // 获取添加拦截器字段名，字段顺序
            timestampSerialNumber = context.getInteger("timestampSerialNumber");
            separatorString = context.getString("separator", "\t");
        }
    }
}
